package com.yuwen.yuwenRPC.server.tcp;

import cn.hutool.core.util.IdUtil;
import com.yuwen.RpcApplication;
import com.yuwen.yuwenRPC.model.RpcRequest;
import com.yuwen.yuwenRPC.model.RpcResponse;
import com.yuwen.yuwenRPC.model.ServiceMetaInfo;
import com.yuwen.yuwenRPC.protocol.*;
import com.yuwen.yuwenRPC.registry.Registry;
import com.yuwen.yuwenRPC.registry.RegistryFactory;
import com.yuwen.yuwenRPC.serializer.Serializer;
import io.vertx.core.Vertx;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.net.NetClient;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

@Slf4j
public class VertxTcpClient {
    /**
     * TCP 请求
     * @param serializer 序列化器
     * @param rpcRequest 请求
     * @param requestServiceMetaInfo 服务元信息
     * @return rpc响应
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public static  RpcResponse tcpRequest(Serializer serializer, RpcRequest rpcRequest, ServiceMetaInfo requestServiceMetaInfo) throws ExecutionException, InterruptedException {
        try {
            byte[] bytes = serializer.serialize(rpcRequest);

            //rcp发送tcp请求
            Vertx vertx = Vertx.vertx();
            NetClient netClient = vertx.createNetClient();
            CompletableFuture<RpcResponse> futureResponse = new CompletableFuture<>();

            netClient.connect(requestServiceMetaInfo.getServicePort(),
                    requestServiceMetaInfo.getServiceHost(),
                    result->{
                        if (result.succeeded()) {
                            log.info("Succeed to connect to server");
                            io.vertx.core.net.NetSocket socket = result.result();
                            //发送数据
                            //构造请求
                            ProtocolMessage<RpcRequest> protocolMessage = new ProtocolMessage<>();
                            ProtocolMessage.Header header = new ProtocolMessage.Header();
                            header.setMagic(ProtocolConstant.PROTOCOL_MAGIC);
                            header.setVersion(ProtocolConstant.PROTOCOL_VERSION);
                            header.setSerializer((byte) ProtocolMessageSerializerEnum.getSerializerByValue(RpcApplication.getRpcConfig().getSerializer()).getKey());
                            header.setType((byte) ProtocolMessageTypeEnum.REQUEST.getKey());
                            header.setRequestId(IdUtil.getSnowflakeNextId());

                            protocolMessage.setHeader(header);
                            protocolMessage.setBody(rpcRequest);
                            try {
                                Buffer buffer = ProtocolMessageEncoder.encode(protocolMessage);
                                socket.write(buffer);
                            } catch (IOException e) {
                                throw new RuntimeException("协议消息编码错误");
                            }
                            //接收响应
                            TCPBufferHandlerWrapper tcpBufferHandlerWrapper= new TCPBufferHandlerWrapper(buffer->{
                                try {
                                    ProtocolMessage<RpcResponse> rpcResponseProtocolMessage = (ProtocolMessage<RpcResponse>) ProtocolMessageDecoder.decode(buffer);
                                    futureResponse.complete(rpcResponseProtocolMessage.getBody());
                                } catch (IOException e) {
                                    throw new RuntimeException("协议消息解码错误");
                                }
                            });
                            socket.handler(tcpBufferHandlerWrapper);
                        }
                        else {
//                            应该删除注册中心对应无效的节点
                            String registryKey = RpcApplication.getRpcConfig().getRegistryConfig().getRegistryType();
                            Registry registry = RegistryFactory.getRegistry(registryKey);
                            try {
                                registry.unRegister(requestServiceMetaInfo);
                            } catch (Exception e) {
                                throw new RuntimeException(e);
                            }
                            log.error("Fail to connect to server");
                            futureResponse.completeExceptionally(new RuntimeException("Fail to connect to server"));
                        }
                    });
            RpcResponse rpcResponse = futureResponse.get();
            //关闭连接
            netClient.close();
            return rpcResponse;
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }

    /**
     * 启动服务方法
     */
    public void start() {
        // 创建 Vert.x 实例
        Vertx vertx = Vertx.vertx();

        vertx.createNetClient().connect(8888, "localhost", result -> {
            if (result.succeeded()) {
                System.out.println("Connected to TCP server");
                io.vertx.core.net.NetSocket socket = result.result();
                for (int i = 0; i < 1000; i++) {
                    String str = "Hello, server!Hello, server!Hello, server!Hello, server!";
                    Buffer buffer = Buffer.buffer();
                    buffer.appendInt(0);
                    buffer.appendInt(str.length());
                    buffer.appendBytes(str.getBytes());
                    socket.write(buffer);
                }
                // 接收响应
                socket.handler(buffer -> {
                    System.out.println("Received response from server: " + buffer.toString());
                });
            } else {
                System.err.println("Failed to connect to TCP server");
            }
        });
    }

    public static void main(String[] args) {
        new VertxTcpClient().start();
    }
}
